<?php

namespace app\queue\redis\download;

use app\extensions\api\Oss;
use app\extensions\lib\RedisLock;
use app\model\TokenInfo;
use OSS\Core\OssException;
use support\Db;
use support\Log;
use support\Redis;
use Webman\RedisQueue\Consumer;

class DownToken implements Consumer
{
    // 要消费的队列名
    public $queue = 'queue_download_tokens';
    public static $redis_queue_key = 'queue_download_tokens';
    public $connection = 'default';

    public static $path = 'admin_download/test_push/';


    public static function handleSql($sql,$bindings){
        $sqls = explode('?',$sql);
        $newSql = '';
        foreach ($sqls as $k => $sqlt){
            $value = $bindings[$k] ?? '';
            if(empty($value)) continue;
            $newSql .= $sqlt.(is_numeric($value) ? $value : '\''.$value.'\'');
        }
        return $newSql;
    }

    public static function getBuild($whereList){
        $build = new TokenInfo();
        foreach ($whereList as $wheres){
            list($operator,$params) = $wheres;
            $build = $build->{$operator}(...$params);
        }
        return $build;
    }

    public static function statusKey($object){
        return  'status__'.$object;
    }

    public static function positionKey($object){
        return  'position__'.$object;
    }


    /***
     * 对外支持分页
     * @param $data
     * @return int|void
     */
    public static function publicSend($data){
        $whereList = $data['whereList'] ?? [];
        $object = $data['object'] ?? '';
        /* @var $build \Illuminate\Database\Eloquent\Builder */
        $build = self::getBuild($whereList);

        $system_limit = TokenInfo::TOKEN_LIMIT;//查询分页限制
        $times = 1;
        if($build->count()>$system_limit){
            $count = $build->count();
            $times = ceil($count/$system_limit);
            Redis::set(self::statusKey($object),$times);//记录次数
            Redis::expire(self::statusKey($object),86400);
            for($i=1;$i<=$times;$i++){
                $new_data = $data;
                $offset = ($i - 1) * $system_limit;
                $new_data['offset'] = $offset;
                $new_data['limit'] = $system_limit;
                $status = \Webman\RedisQueue\Redis::send(self::$redis_queue_key,$new_data,$i);
            }
        }else{
            Redis::set(self::statusKey($object),1);//记录次数
            Redis::expire(self::statusKey($object),86400);//记录次数
            \Webman\RedisQueue\Redis::send(self::$redis_queue_key,$data);
        }

        return $times;


    }





    // 消费
    public function consume($data)
    {

        $whereList = $data['whereList'] ?? [];
        /* @var $build \Illuminate\Database\Eloquent\Builder */
        $build = $this->getBuild($whereList);
        $data['build_data'] = [$build->toSql(),$build->getBindings()];
        $object = $data['object'] ?? '';
        $field = $data['field'] ?? '';
        $offset = $data['offset'] ?? 0;
        $limit = $data['limit'] ?? 0;
        $position_key = self::positionKey($object);
        if(!$build || !$object || !$field){
            Log::channel('test')->warning('no build or object or field:'.json_encode($data));
            return;
        }
        $times = Redis::get(self::statusKey($object));
        if(!$times){
            Log::channel('test')->warning('no记录次数:'.json_encode($data));
        }
        //是否限制
        if($limit){
            $build = $build->offset($offset)->limit($limit);
        }

        $system_limit = TokenInfo::TOKEN_LIMIT;//查询分页限制
        $list = $build->select($field)->get()->toArray();
        $count = count($list);
        if(!$count){
            Log::channel('test')->warning("count:$count,no udid in token_infos-".$system_limit.':'.json_encode($data));
            //return;
        }
        if($count>$system_limit){
            Log::channel('test')->warning($count.'=build->select()->get() > '.$system_limit.':'.json_encode($data));
            return;
        }
        $lists = array_chunk($list,1000);//阿里云上传内容限制
        $contents = [];
        foreach ($lists as $items){
            $content = '';
            $title = implode(',',$field)."\r\n";
            foreach ($items as $item) {
                $content .= implode(',',array_values($item))."\r\n";
            }
            $contents[] = $content;
        }
        if(empty($contents)) $contents = [''];//空也需要写入文件


        $param = [
            'data' => json_encode($data),
        ];
        $callbackUrl = env('APP_URL').'oss/download_oss_callback';
        $Oss = new Oss();
        //$url = Oss::DOMAIN.'/'.$object;
        $RedisLock = new RedisLock();
        $lockKey = 'lock__'.$object;
        //使用悲观锁
        $position = $RedisLock->actionLock($lockKey,function () use ($Oss,$contents,$object,$param,$callbackUrl,$position_key){
            try{
                $position = Redis::get($position_key) ?: 0;//重新获取位置
                $position  = $Oss->uploadContentsCallback($contents,$object,$param,$callbackUrl,$position);
                Redis::set($position_key,$position);//下载限制
                Redis::expire($position_key,86400);
            }catch (OssException $ossException){
                Log::channel('test')->error('uploadContentsCallback==OssException:'.$object.':'.$ossException->getMessage());
            }

            return $position ?? -1;
        });
        //成功递减次数
        $incr_status = Redis::incr(self::statusKey($object),-1);
        Log::channel('test')->info("object:{$object},count:$count,position:{$position},incr_status:{$incr_status}".json_encode($data));


    }
}